package com.atguigu.realtime.dwd.db.app;

import com.atguigu.gmall.realtime.common.base.BaseSQLApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeCartAdd extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdTradeCartAdd().start(
            10013,
            4,
            Constant.TOPIC_DWD_TRADE_CART_ADD
        );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tableEnv, String groupId) {

        createTopicDb(groupId, tableEnv);

//        Table cartAddTable = tableEnv.sqlQuery("select\n" +
//                "`data`['id'] id,\n" +
//                "`data`['user_id'] user_id,\n" +
//                "`data`['sku_id'] sku_id,\n" +
//                "`data`['cart_price'] cart_price,\n" +
//                "if(`type` = 'insert', `data`['sku_num'], cast(cast(`data`['sku_num'] as bigint ) - cast(`old`['sku_num'] as bigint)as string )) sku_num,\n" +
//                "`data`['sku_name'] sku_name,\n" +
//                "`data`['is_checked'] is_checked,\n" +
//                "`data`['create_time'] create_time,\n" +
//                "`data`['operate_time'] operate_time,\n" +
//                "`data`['is_ordered'] is_ordered,\n" +
//                "`data`['order_time'] order_time,\n" +
//                "`data`['source_type'] source_type,\n" +
//                "`data`['source_id'] source_id,\n" +
//                "ts\n" +
//                "from topic_db \n" +
//                "where `database` = 'gmall'\n" +
//                "and `table` = 'cart_info'\n" +
//                "and (`type` = 'insert'  or (\n" +
//                "`type` = 'update' and `old`['sku_num'] is not null \n" +
//                "and cast(`data`['sku_num'] as bigint ) > cast(`old`['sku_num'] as bigint)))");


/// 2. 过滤出加购数据
        Table cartAdd = tableEnv.sqlQuery(
                "select " +
                        " `data`['id'] id," +
                        " `data`['user_id'] user_id," +
                        " `data`['sku_id'] sku_id," +
                        " if(`type`='insert'," +
                        "   cast(`data`['sku_num'] as int), " +
                        "   cast(`data`['sku_num'] as int) - cast(`old`['sku_num'] as int)" +
                        ") sku_num ," +
                        " ts " +
                        "from topic_db " +
                        "where `database`='gmall' " +
                        "and `table`='cart_info' " +
                        "and (" +
                        " `type`='insert' " +
                        "  or(" +
                        "     `type`='update' " +
                        "      and `old`['sku_num'] is not null " +
                        "      and cast(`data`['sku_num'] as int) > cast(`old`['sku_num'] as int) " +
                        "   )" +
                        ")");

        // 3. 写出到 kafka
        tableEnv.executeSql("create table dwd_trade_cart_add(" +
                "   id string, " +
                "   user_id string," +
                "   sku_id string," +
                "   sku_num int, " +
                "   ts  bigint " +
                ")" + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_CART_ADD));

        cartAdd.executeInsert(Constant.TOPIC_DWD_TRADE_CART_ADD);
    }
}